#include "index.h"
#include "kvstore.h"
#include "fingerprint_cache.h"
#include "../storage/containerstore.h"
#include "../recipe/recipestore.h"
#include "../jcr.h"
#include "../destor.h"
#include "../utils/rabin_chunking.h"
#include "db_mysql.h"
#include "../utils/bloom_filter.h"

/* The buffer size > 2 * destor.rewrite_buffer_size */
/* All fingerprints that have been looked up in the index
 * but not been updated. */

struct
{
    /* map a fingerprint to a queue of indexElem */
    /* Index all fingerprints in the index buffer. */
    GHashTable *buffered_fingerprints;
    /* The number of buffered chunks */
    int chunk_num;
} index_buffer;

struct
{
    /* Requests to the key-value store */
    int lookup_requests;
    int update_requests;
    int lookup_requests_for_unique;
    /* Overheads of prefetching module */
    int read_prefetching_units;
} index_overhead;

// FILE *excel_similarity;
#define min(a, b) ((a) < (b) ? (a) : (b))

static unsigned char *filter;
static int dirty = FALSE;
int64_t index_memory_overhead = 0;

gboolean g_feature_equal(char *a, char *b)
{
    return !memcmp(a, b, destor.index_key_size);
}

guint g_feature_hash(char *feature)
{
    int i, hash = 0;
    for (i = 0; i < destor.index_key_size; i++)
    {
        hash += feature[i] << (8 * i);
    }
    return hash;
}

void init_index()
{
    // excel_similarity = fopen("excel_similarity.txt", "a");
    /* Do NOT assign a free function for value. */
    index_buffer.buffered_fingerprints = g_hash_table_new_full(g_int64_hash,
                                                               g_fingerprint_equal, NULL, NULL);
    index_buffer.chunk_num = 0;
    /*
    if (db_init() == FALSE)
    {
        return FALSE;
    }
    */
    sds indexpath = sdsdup(destor.working_directory);
    indexpath = sdscat(indexpath, "index/bloom_filter");
    int fd;
    if ((fd = open(indexpath, O_RDONLY | O_CREAT, S_IRWXU)) <= 0)
    {
        printf("Can not open index/bloom_filter!");
        return FALSE;
    }
    filter = malloc(FILTER_SIZE_BYTES);
    if (FILTER_SIZE_BYTES != read(fd, filter, FILTER_SIZE_BYTES))
    {
        bzero(filter, FILTER_SIZE_BYTES);
    }
    close(fd);
    dirty = FALSE;

    if (destor.index_specific != INDEX_SPECIFIC_NO)
    {
        destor.index_key_size = sizeof(fingerprint);
        switch (destor.index_specific)
        {
        case INDEX_SPECIFIC_DDFS:
        {
            destor.index_category[0] = INDEX_CATEGORY_EXACT;
            destor.index_category[1] = INDEX_CATEGORY_PHYSICAL_LOCALITY;
            break;
        }
        case INDEX_SPECIFIC_BLOCK_LOCALITY_CACHING:
        {
            destor.index_category[0] = INDEX_CATEGORY_EXACT;
            destor.index_category[1] = INDEX_CATEGORY_LOGICAL_LOCALITY;
            destor.index_sampling_method[0] = INDEX_SAMPLING_UNIFORM;
            destor.index_sampling_method[1] = 1;

            destor.index_segment_prefech = destor.index_segment_prefech > 1 ? destor.index_segment_prefech : 16;
            break;
        }
        case INDEX_SPECIFIC_SAMPLED:
        {
            destor.index_category[0] = INDEX_CATEGORY_NEAR_EXACT;
            destor.index_category[1] = INDEX_CATEGORY_PHYSICAL_LOCALITY;

            destor.index_sampling_method[0] = INDEX_SAMPLING_UNIFORM;
            destor.index_sampling_method[1] = destor.index_sampling_method[1] > 1 ? destor.index_sampling_method[1] : 128;
            break;
        }
        case INDEX_SPECIFIC_SPARSE:
        {
            destor.index_category[0] = INDEX_CATEGORY_NEAR_EXACT;
            destor.index_category[1] = INDEX_CATEGORY_LOGICAL_LOCALITY;

            destor.index_segment_algorithm[0] = INDEX_SEGMENT_CONTENT_DEFINED;

            destor.index_segment_selection_method[0] = INDEX_SEGMENT_SELECT_TOP;

            destor.index_sampling_method[0] = INDEX_SAMPLING_RANDOM;
            destor.index_sampling_method[1] = destor.index_sampling_method[1] > 1 ? destor.index_sampling_method[1] : 128;

            destor.index_segment_prefech = 1;
            break;
        }
        case INDEX_SPECIFIC_SILO:
        {
            destor.index_category[0] = INDEX_CATEGORY_NEAR_EXACT;
            destor.index_category[1] = INDEX_CATEGORY_LOGICAL_LOCALITY;

            destor.index_segment_algorithm[0] = INDEX_SEGMENT_FIXED;

            destor.index_segment_selection_method[0] = INDEX_SEGMENT_SELECT_TOP;
            destor.index_segment_selection_method[1] = 1;

            destor.index_sampling_method[0] = INDEX_SAMPLING_MIN;
            destor.index_sampling_method[1] = 0;

            destor.index_segment_prefech = destor.index_segment_prefech > 1 ? destor.index_segment_prefech : 16;
            break;
        }
        default:
        {
            WARNING("Invalid index specific!");
            exit(1);
        }
        }
    }

    if (destor.index_category[0] == INDEX_CATEGORY_EXACT)
        destor.index_key_size = sizeof(fingerprint);

    if (destor.index_category[1] == INDEX_CATEGORY_PHYSICAL_LOCALITY)
    {
        // destor.index_segment_algorithm[0] = INDEX_SEGMENT_FIXED;
        if (destor.index_category[0] == INDEX_CATEGORY_EXACT)
        {
            destor.index_sampling_method[0] = INDEX_SAMPLING_UNIFORM;
            destor.index_sampling_method[1] = 1;
        }
    }
    else
        assert(destor.enable_sidc == 0);

    assert(destor.index_key_size > 0 && destor.index_key_size <= sizeof(fingerprint));

    init_sampling_method();
    init_segmenting_method();

    init_kvstore();

    init_fingerprint_cache();

    index_overhead.lookup_requests = 0;
    index_overhead.update_requests = 0;
    index_overhead.lookup_requests_for_unique = 0;
    index_overhead.read_prefetching_units = 0;
}

void close_index()
{
    close_kvstore();
    full_index_destroy();
    free_fingerprint_cache();
}

extern struct
{
    /* accessed in dedup phase */
    struct container *container_buffer;
    /* In order to facilitate sampling in container,
     * we keep a queue for chunks in container buffer. */
    GSequence *chunks;
} storage_buffer;

extern void index_lookup_similarity_detection(struct segment *s);

extern struct
{
    /* g_mutex_init() is unnecessary if in static storage. */
    GMutex mutex;
    GCond cond; // index buffer is not full
    // index buffer is full, waiting
    // if threshold < 0, it indicates no threshold.
    int wait_threshold;
} index_lock;

// struct
// {
//     /* accessed in dedup phase */
//     struct container *container_buffer;
//     /* In order to facilitate sampling in container,
//      * we keep a queue for chunks in container buffer. */
//     GSequence *chunks;
// } storage_buffer;


// struct
// {
//     /* g_mutex_init() is unnecessary if in static storage. */
//     GMutex mutex;
//     GCond cond; // index buffer is not full
//     // index buffer is full, waiting
//     // if threshold < 0, it indicates no threshold.
//     int wait_threshold;
// } index_lock;

void lookup_base(struct segment *s)
{
    GSequenceIter *iter = g_sequence_get_begin_iter(s->chunks);
    GSequenceIter *end = g_sequence_get_end_iter(s->chunks);
    for (; iter != end; iter = g_sequence_iter_next(iter))
    {
        struct chunk *c = g_sequence_get(iter);

        if (CHECK_CHUNK(c, CHUNK_FILE_START) || CHECK_CHUNK(c, CHUNK_FILE_END))
            continue;
        /* First check it in the storage buffer */

        if (storage_buffer.container_buffer && lookup_chunk_in_container(storage_buffer.container_buffer, c))
        {
            c->id = get_container_id(storage_buffer.container_buffer);
            SET_CHUNK(c, CHUNK_DUPLICATE);
            SET_CHUNK(c, CHUNK_REWRITE_DENIED);
        }

        /*
         * Then check the buffered fingerprints,
         * recently backup fingerprints.
         */

        GQueue *tq = g_hash_table_lookup(index_buffer.buffered_fingerprints, &c->fp);
        if (!tq)
        {
            tq = g_queue_new();
        }
        else if (!CHECK_CHUNK(c, CHUNK_DUPLICATE))
        {
            /* chunk exist but is not duplicate */
            struct indexElem *be = g_queue_peek_head(tq);
            c->id = be->id;
            SET_CHUNK(c, CHUNK_DUPLICATE);
            if (c->id == TEMPORARY_ID)
                c->type = CHUNK_TYPE_DUPLICATE_ALL; // The buffered fingerprints have not the segment recipes
            else
            {
                if (be->type == CHUNK_TYPE_UNIQUE || be->type == CHUNK_TYPE_DUPLICATE)
                    c->type = CHUNK_TYPE_DUPLICATE;
                else
                    c->type = CHUNK_TYPE_DUPDELTA;
            }
        }

        /*
         *  Check the fingerprint cache
         */
        if (!CHECK_CHUNK(c, CHUNK_DUPLICATE))
        {
            /* Searching in fingerprint cache */
            int64_t id = delta_fingerprint_cache_lookup(c);

            if (id != TEMPORARY_ID)
            {
                c->id = id;
                SET_CHUNK(c, CHUNK_DUPLICATE);
            }
        }

        /*
         *  Finally, check the key-value store.
         */
        if (!CHECK_CHUNK(c, CHUNK_DUPLICATE))
        {
            /*
            if (in_dict(filter, (char *)&c->fp, sizeof(fingerprint)))
            {
                containerid cid = db_lookup_fingerprint(&c->fp);

                if (cid != TEMPORARY_ID)
                {
                    index_overhead.lookup_requests++;

                    index_overhead.read_prefetching_units++;

                    fingerprint_cache_prefetch(cid);

                    int64_t id = delta_fingerprint_cache_lookup(c);
                    if (id != TEMPORARY_ID)
                    {
                        c->id = id;
                        SET_CHUNK(c, CHUNK_DUPLICATE);
                    }
                    else
                    {
                        NOTICE("Filter phase: A key collision occurs");
                    }
                }
            }
            */

            int64_t *ids = kvstore_lookup((char *)&c->fp);
            if (ids)
            {

                fingerprint_cache_prefetch(ids[0]);

                int64_t id = delta_fingerprint_cache_lookup(c);

                if (id != TEMPORARY_ID)
                {
                    c->id = id;
                    SET_CHUNK(c, CHUNK_DUPLICATE);
                }
                else
                {
                    NOTICE("Filter phase: A key collision occurs");
                }
            }
            else
            {
                index_overhead.lookup_requests_for_unique++;
                VERBOSE("Dedup phase: non-existing fingerprint");
            }
        }

        if (!CHECK_CHUNK(c, CHUNK_DUPLICATE))
            c->type = CHUNK_TYPE_UNIQUE_ALL;

        /* Insert it into the index buffer */
        struct indexElem *ne = (struct indexElem *)malloc(sizeof(struct indexElem));
        ne->id = c->id;
        ne->type = c->type;
        memcpy(&ne->fp, &c->fp, sizeof(fingerprint));

        g_queue_push_tail(tq, ne);
        g_hash_table_replace(index_buffer.buffered_fingerprints, &ne->fp, tq);

        index_buffer.chunk_num++;
    }
}

/*
 * return 1: indicates lookup is successful.
 * return 0: indicates the index buffer is full.
 */
int index_lookup(struct segment *s)
{

    /* Ensure the next phase not be blocked. */
    if (index_lock.wait_threshold > 0 && index_buffer.chunk_num >= index_lock.wait_threshold)
    {
        DEBUG("The index buffer is full (%d chunks in buffer)",
              index_buffer.chunk_num);
        return 0;
    }

    TIMER_DECLARE(1);
    TIMER_BEGIN(1);
    // s->features = sampling(s->chunks, s->chunk_num);
    // index_lookup_similarity_detection(s);//效果更好，但是cds效果下降
    lookup_base(s);
    TIMER_END(1, jcr.dedup_time);

    return 1;
}

/*
 * Input features with an ID.
 */
void index_update(GHashTable *features, int64_t id)
{
    // NOTICE("Filter phase: update %d features", g_hash_table_size(features));
    GHashTableIter iter;
    gpointer key, value;
    g_hash_table_iter_init(&iter, features);
    while (g_hash_table_iter_next(&iter, &key, &value))
    {
        index_overhead.update_requests++;
        kvstore_update(key, id);
        // db_insert_fingerprint(key, id);
        // insert_word(filter, (char *)key, sizeof(fingerprint));
        dirty = TRUE;
    }
}

inline void index_delete(fingerprint *fp, int64_t id)
{
    // kvstore_delete(fp, id);
}

/* This function is designed for rewriting. */
void index_check_buffer(struct segment *s)
{

    GSequenceIter *iter = g_sequence_get_begin_iter(s->chunks);
    GSequenceIter *end = g_sequence_get_end_iter(s->chunks);
    for (; iter != end; iter = g_sequence_iter_next(iter))
    {
        struct chunk *c = g_sequence_get(iter);

        if (CHECK_CHUNK(c, CHUNK_FILE_START) || CHECK_CHUNK(c, CHUNK_FILE_END))
            continue;

        if ((CHECK_CHUNK(c, CHUNK_DUPLICATE) && c->id == TEMPORARY_ID) || CHECK_CHUNK(c, CHUNK_OUT_OF_ORDER) || CHECK_CHUNK(c, CHUNK_SPARSE) || c->DepWithDelta)
        {
            /*
             * 1. A duplicate chunk with a TEMPORARY_ID,
             * 	  indicates it is a copy of a recent unique chunk.
             * 2. If we want to rewrite a chunk,
             *    we check whether it has been rewritten recently.
             */
            GQueue *bq = g_hash_table_lookup(index_buffer.buffered_fingerprints, &c->fp);
            assert(bq);
            struct indexElem *be = g_queue_peek_head(bq);

            if (be->id > c->id)
            {

                if (c->id != TEMPORARY_ID)
                {
                    /* this chunk has been rewritten recently
                     * since: 1) the rewritten chunks exist in index_buffer
                     *        2) the rewritten chunks have containerid */
                    SET_CHUNK(c, CHUNK_REWRITE_DENIED);
                }

                c->id = be->id;
                // c->baseid = be->baseid;
                c->DepWithDelta = be->DepWithDelta;
                // c->base_size = be->base_size;
                // c->delta_size = be->delta_size;
                if (be->type == CHUNK_TYPE_UNIQUE || c->type == CHUNK_TYPE_DUPLICATE)
                    c->type = CHUNK_TYPE_DUPLICATE;
                else
                    c->type = CHUNK_TYPE_DUPDELTA;
            }
        }
    }
}

/*
 * Return 1 indicates buffer remains full.
 */
int index_update_buffer(struct segment *s)
{
    GSequenceIter *iter = g_sequence_get_begin_iter(s->chunks);
    GSequenceIter *end = g_sequence_get_end_iter(s->chunks);
    for (; iter != end; iter = g_sequence_iter_next(iter))
    {
        struct chunk *c = g_sequence_get(iter);

        if (CHECK_CHUNK(c, CHUNK_FILE_START) || CHECK_CHUNK(c, CHUNK_FILE_END))
            continue;

        assert(c->id != TEMPORARY_ID);
        // assert(c->type != CHUNK_TYPE_DUPLICATE_ALL && c->type != CHUNK_TYPE_UNIQUE_ALL);

        GQueue *bq = g_hash_table_lookup(index_buffer.buffered_fingerprints, &c->fp);

        struct indexElem *e = g_queue_pop_head(bq);
        if (g_queue_get_length(bq) == 0)
        {
            g_hash_table_remove(index_buffer.buffered_fingerprints, &c->fp);
            g_queue_free(bq);
        }
        else
        {
            int num = g_queue_get_length(bq), j;
            for (j = 0; j < num; j++)
            {
                /*
                 * set all container id in this buck equal to the cid of the first stored,
                 * note that if the current c is not the first stored chunk, it's cid will equals
                 * to the first stored chunk due to the recently_rewritten_chunks in filter_phase
                 */
                struct indexElem *ie = g_queue_peek_nth(bq, j);
                ie->id = c->id;
                // ie->baseid = c->baseid;
                ie->DepWithDelta = c->DepWithDelta;
                // ie->base_size = c->base_size;
                // ie->delta_size = c->delta_size;

                if (c->deltaCompressed || c->DepWithDelta || c->type == CHUNK_TYPE_DELTA || c->type == CHUNK_TYPE_DUPDELTA)
                {
                    ie->DepWithDelta = 1;
                    ie->type = CHUNK_TYPE_DUPDELTA;
                }
                else
                {
                    ie->type = CHUNK_TYPE_DUPLICATE;
                }
            }
        }
        free(e);
        index_buffer.chunk_num--;
    }

    if (index_lock.wait_threshold <= 0 || index_buffer.chunk_num < index_lock.wait_threshold)
    {
        DEBUG("The index buffer is ready for more chunks (%d chunks in buffer)",
              index_buffer.chunk_num);
        return 0;
    }
    return 1;
}

void prinfChunkState(struct chunk *c)
{
    printf("duplicate: %d, sparse: %d, in cache: %d, deny: %d, fragment: %d, base in cache: %d\n",
           CHECK_CHUNK(c, CHUNK_DUPLICATE), CHECK_CHUNK(c, CHUNK_SPARSE),
           CHECK_CHUNK(c, CHUNK_IN_CACHE), CHECK_CHUNK(c, CHUNK_REWRITE_DENIED),
           CHECK_CHUNK(c, CHUNK_FRAGMENT), CHECK_CHUNK(c, CHUNK_BASE_IN_CACHE));
    printf("c->id: %lld, c->baseid: %lld, c->delta: %s\n",
           c->id, c->baseid, c->delta ? "yes" : "no");
}

void bloom_filter_flush()
{
    if (dirty == FALSE)
        return;
    /* flush bloom filter */
    sds indexpath = sdsdup(destor.working_directory);
    indexpath = sdscat(indexpath, "index/bloom_filter");

    int fd;
    if ((fd = open(indexpath, O_WRONLY | O_CREAT, S_IRWXU)) <= 0)
    {
        printf("Can not open index/bloom_filter!");
    }
    if (FILTER_SIZE_BYTES != write(fd, filter, FILTER_SIZE_BYTES))
    {
        printf("%s, %d: Failed to flush bloom filter!\n", __FILE__, __LINE__);
    }
    close(fd);

    dirty = FALSE;
    sdsfree(indexpath);
}

void full_index_destroy()
{
    // index_memory_overhead = db_close(); // one byte for each chunk
    bloom_filter_flush();
    free(filter);
}

void full_index_update(fingerprint *finger, containerid id)
{
    db_insert_fingerprint(finger, id);

    insert_word(filter, (char *)finger, sizeof(fingerprint));
    dirty = TRUE;
}
